package hy;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class MyProducer {
    public static void main(String[] args) throws Exception {
        Properties p=new Properties();
        p.load(MyProducer.class.getClassLoader().getResourceAsStream("producer.properties"));
        
        Producer<String,String> pr=new KafkaProducer<>(p);
        for(int i=0;i<10;i++){
            //pr.send(new ProducerRecord("first","hello_"+i));
            //pr.send(new ProducerRecord("first","abc","Hello" + i));
            pr.send(new ProducerRecord("first", "Hello" + i), (m, e) -> {
                //收到ack是调用，发送失败且不再重发也会调用该方法
                if(e==null){
                    //发送成功
                    System.out.println("分区："+m.partition()+"----offset:"+m.offset());
                }else {
                    System.out.println("发送失败");
                }
            }).get();// 注意：调用get的后果就是让发送方法阻塞，当收到ack并调用了回调后就解除阻塞。
        }
        pr.close();
    }
}